package com.cc.flink.config;

import java.util.Properties;

/**
 * 作业配置管理类
 * 集中管理Flink作业的所有配置参数，实现配置外部化和默认值管理
 * 采用单例模式设计，确保配置的一致性和可维护性
 */
public class JobConfig {
    // Kafka相关配置参数
    private final String kafkaBootstrapServers; // Kafka集群地址列表
    private final String kafkaGroupId;         // 消费者组ID
    private final String kafkaTopic;           // 要消费的主题名称

    // Doris数据库相关配置
    private final String dorisUrl;             // Doris JDBC连接URL
    private final String dorisUsername;        // Doris用户名
    private final String dorisPassword;        // Doris密码
    private final String dorisTable;           // 目标表名

    // 流式处理窗口配置
    private final long windowSizeSeconds;      // 窗口大小（秒）
    private final long maxRecordsPerBatch;     // 每批次最大记录数，用于批量写入

    // Flink检查点配置，用于保证容错和状态恢复
    private final long checkpointIntervalMs;   // 检查点间隔（毫秒）
    private final String checkpointPath;       // 检查点保存路径
    private final int maxConcurrentCheckpoints; // 最大并发检查点数
    private final long checkpointTimeoutMs;    // 检查点超时时间（毫秒）

    /**
     * 构造函数
     * 初始化配置，优先从环境变量读取，环境变量不存在时使用默认值
     * 这种设计支持运行时动态调整配置，无需修改代码
     */
    public JobConfig() {
        // Kafka配置
        this.kafkaBootstrapServers = getEnv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092");
        this.kafkaGroupId = getEnv("KAFKA_GROUP_ID", "flink-kafka-doris-group");
        this.kafkaTopic = getEnv("KAFKA_TOPIC", "input-topic");

        // Doris配置
        this.dorisUrl = getEnv("DORIS_URL", "jdbc:mysql://localhost:9030/demo_db");
        this.dorisUsername = getEnv("DORIS_USERNAME", "root");
        this.dorisPassword = getEnv("DORIS_PASSWORD", "password");
        this.dorisTable = getEnv("DORIS_TABLE", "target_table");

        // 窗口和批处理配置
        this.windowSizeSeconds = Long.parseLong(getEnv("WINDOW_SIZE_SECONDS", "60"));
        this.maxRecordsPerBatch = Long.parseLong(getEnv("MAX_RECORDS_PER_BATCH", "100"));

        // 检查点配置
        this.checkpointIntervalMs = Long.parseLong(getEnv("CHECKPOINT_INTERVAL_MS", "60000"));
        this.checkpointPath = getEnv("CHECKPOINT_PATH", "file:///tmp/flink/checkpoints");
        this.maxConcurrentCheckpoints = Integer.parseInt(getEnv("MAX_CONCURRENT_CHECKPOINTS", "1"));
        this.checkpointTimeoutMs = Long.parseLong(getEnv("CHECKPOINT_TIMEOUT_MS", "600000"));
    }

    /**
     * 获取环境变量值的工具方法
     *
     * @param key          环境变量名称
     * @param defaultValue 当环境变量不存在时返回的默认值
     * @return 环境变量值或默认值
     */
    private String getEnv(String key, String defaultValue) {
        String value = System.getenv(key);
        return value != null ? value : defaultValue;
    }

    /**
     * 获取Kafka连接属性配置
     * 将内部配置转换为标准的Properties格式，便于创建Kafka消费者
     *
     * @return Kafka连接属性
     */
    public Properties getKafkaProperties() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", kafkaBootstrapServers);
        props.setProperty("group.id", kafkaGroupId);
        return props;
    }

    /**
     * 生成Doris数据库插入SQL语句
     * 动态生成SQL，支持不同的表名配置
     *
     * @return 预编译的插入SQL语句
     */
    public String getDorisInsertSql() {
        return String.format("INSERT INTO %s (id, data) VALUES (?, ?)", dorisTable);
    }

    // Getter方法
    public String getKafkaBootstrapServers() {
        return kafkaBootstrapServers;
    }

    public String getKafkaGroupId() {
        return kafkaGroupId;
    }

    public String getKafkaTopic() {
        return kafkaTopic;
    }

    public String getDorisUrl() {
        return dorisUrl;
    }

    public String getDorisUsername() {
        return dorisUsername;
    }

    public String getDorisPassword() {
        return dorisPassword;
    }

    public String getDorisTable() {
        return dorisTable;
    }

    public long getWindowSizeSeconds() {
        return windowSizeSeconds;
    }

    /**
     * 获取批次窗口时间（毫秒）
     */
    public long getBatchWindowTimeMs() {
        return getWindowSizeSeconds() * 1000L;
    }

    public long getMaxRecordsPerBatch() {
        return maxRecordsPerBatch;
    }

    public long getCheckpointIntervalMs() {
        return checkpointIntervalMs;
    }

    public String getCheckpointPath() {
        return checkpointPath;
    }

    public int getMaxConcurrentCheckpoints() {
        return maxConcurrentCheckpoints;
    }

    public long getCheckpointTimeoutMs() {
        return checkpointTimeoutMs;
    }
}